
package net.bwie.vehicle.dws.function;

import net.bwie.vehicle.dws.bean.CarBean;
import net.bwie.vehicle.dws.bean.TripDetectionResult;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * JourneyRealtimeFunction是一个用于实时计算行程信息的 keyed process function.
 * 它主要通过车辆的速度来判断行程的开始和结束，并计算行程的相关信息，如开始时间、结束时间和行驶距离.
 */
public class JourneyRealtimeFunction extends KeyedProcessFunction<String, CarBean, TripDetectionResult> {



    // 用于存储行程开始的时间
    private ValueState<Long> tripStartTimeState;
    // 用于存储行程开始时的里程数
    private ValueState<Double> startMileageState;

    /**
     * 初始化状态.
     * 
     * @param conf 配置参数
     */
    @Override
    public void open(Configuration conf) {
        // 初始化行程开始时间的状态
        tripStartTimeState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("tripStartTime", Long.class));
        // 初始化行程开始时里程数的状态
        startMileageState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("startMileage", Double.class));
    }

    /**
     * 处理每个元素.
     * 
     * @param data 输入的数据
     * @param ctx 上下文信息
     * @param out 用于输出结果的收集器
     * @throws Exception 如果处理过程中出现异常
     */
    @Override
    public void processElement(CarBean data, Context ctx, Collector<TripDetectionResult> out) throws Exception {
        // 基本数据校验
        if (data.getSpeed() == null || data.getTimestamp() == null || data.getVin() == null) {
            return;
        }

        // 简化逻辑：只要速度>0就认为是行程开始
        if (data.getSpeed() > 0) {
            // 如果当前没有记录行程开始时间，则认为是一个新行程的开始
            if (tripStartTimeState.value() == null) {
                // 更新行程开始时间和开始时的里程数
                tripStartTimeState.update(data.getTimestamp());
                startMileageState.update(data.getMileage());
            }
        }
        // 简化逻辑：只要速度=0就认为是行程结束
        else {
            // 如果当前有记录行程开始时间，则认为行程结束
            if (tripStartTimeState.value()>=30000) {
                // 创建并填充行程结果对象
                TripDetectionResult result = new TripDetectionResult();
                // 设置结果中的车架号
                result.setVin(data.getVin());
                // 获取行程ID
                result.setTripId(data.getVin() + "-" + tripStartTimeState.value());
                // 获取结果开始时间
                result.setStartTime(tripStartTimeState.value());
                // 设置结果的结束时间
                result.setEndTime(data.getTimestamp());
                // 计算并设置行驶距离，通过当前里程数减去起始里程数得到
                result.setDistance(data.getMileage() - startMileageState.value());

                // 输出结果
                out.collect(result);

                // 清除状态，为下一段行程做准备
                tripStartTimeState.clear();
                startMileageState.clear();
            }
        }
    }
}
